package org.xian.aio;

import java.io.IOException;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;

/**
 * @author : xian
 */
public class AIOServerHandler implements CompletionHandler<AsynchronousSocketChannel, AIOServer> {


    /** 新的连接进来 调用 */
    @Override
    public void completed(AsynchronousSocketChannel asyncChannel, AIOServer aioServer) {
        /** 再次注册 accept 事件回调，使得服务端可以接收新的客户端连接 */
        aioServer.getAsyncServerSocketChannel().accept(aioServer, this);
        try {
            // 一些设置 先不管
            asyncChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
            asyncChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
            asyncChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
            asyncChannel.setOption(StandardSocketOptions.SO_RCVBUF, 1024);
            asyncChannel.setOption(StandardSocketOptions.SO_SNDBUF, 1024);
        } catch (IOException e) {
            e.printStackTrace();
        }

        read(asyncChannel);
    }

    private void read(AsynchronousSocketChannel asyncChannel) {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        // asyncChannel.read 只是向系统注册一次回调， 这次回调完成后还需要继续向操作系统注册新的 read回调
        asyncChannel.read(buffer, buffer, new CompletionHandler<>() {
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                // 读取客户端的消息
                attachment.flip();
                byte[] bytes = new byte[attachment.remaining()];
                attachment.get(bytes);
                String message = new String(bytes, StandardCharsets.UTF_8);
                try {
                    System.out.println("Aio Server Received --> " + message + ", From --> " + asyncChannel.getRemoteAddress());
                } catch (IOException e) {
                    e.printStackTrace();
                }
                // 简单处理 直接pong回去
                attachment.rewind();
                try {
                    asyncChannel.write(attachment).get();
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
                attachment.clear();
                if ("bye".equals(message)) {
                    try {
                        asyncChannel.shutdownInput();
                        asyncChannel.shutdownOutput();
                        asyncChannel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                } else {
                    // 读取完成后再注册一下回调，有新的消息过来的时候继续由系统调用上面的代码进行处理
                    asyncChannel.read(attachment, attachment, this);
                }
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                exc.printStackTrace();
            }
        });
    }


    @Override
    public void failed(Throwable exc, AIOServer attachment) {
        exc.printStackTrace();
    }
}
